from datetime import datetime, timedelta, date
from airflow import DAG

from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.telegram.operators.telegram import TelegramOperator

import psycopg2
import pandas as pd
import sqlalchemy
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from airflow.models.xcom import XCom

import pendulum

import requests
from bs4 import BeautifulSoup
import logging
import re
import json

LOCAL_TZ = pendulum.timezone("Europe/Moscow")
# ------------------------------------------------------------------------------
# креды для Telegram
# ------------------------------------------------------------------------------
TG_BOT_TOKEN = '7182058516:AAF_Cd8CXft4tcEkfUmWR6kXzEOrGF7FE3w'
TG_CHAT_ID = '-4185266150'
# ------------------------------------------------------------------------------
# Конфиг с параметрами
# ------------------------------------------------------------------------------
with open('config/config.json') as f:
    config_json = json.load(f)
    conn_string = config_json['db_local_string']
    lot_data_columns = config_json['lot_data_columns']
    search_params = config_json['search_params']
# ------------------------------------------------------------------------------
# SQL
# ------------------------------------------------------------------------------
q_stats = """
select
    t1.*,
    t2.lot_id as lot_id_max_price,
    t3.lot_id as lot_id_min_price
from
    (
    select
        (now() - interval '1' day)::date as yesterday_dt,
        sum(case when action_flg = 'I' then 1 else 0 end) as new_lots_cnt,
        sum(case when (action_flg = 'I' and array_length(oks_num, 1) < 2 and array_length(square, 1) = 1 
    and address_formatted is not null and is_share = 0) then 1 else 0 end) as new_lots_valid_cnt,
        sum(case when action_flg = 'U' then 1 else 0 end) as updated_lots_cnt,
        sum(case when action_flg = 'D' then 1 else 0 end) as deleted_lots_cnt,
        max(case when action_flg = 'I' and array_length(oks_num, 1) < 2 and array_length(square, 1) = 1
    and address_formatted is not null and is_share = 0 then initial_price end) as lot_max_price,
        min(case when action_flg = 'I' 
    and array_length(oks_num, 1) < 2 
    and array_length(square, 1) = 1
    and address_formatted is not null
    and is_share = 0 then initial_price end) as lot_min_price
    from
        etl.lot_data ld
    left join etl.lot_data_info ltd
            using(id)
    where
        ld.create_dttm::date = (now() - interval '1' day)::date
    group by
        1 ) t1
left join (
    select
        lot_id,
        initial_price,
        create_dttm::date as dt
    from
        etl.lot_data ld
    where
        action_flg = 'I') t2 on
    t1.yesterday_dt = t2.dt
    and t1.lot_max_price = t2.initial_price
left join (
    select
        lot_id,
        initial_price,
        create_dttm::date as dt
    from
        etl.lot_data ld
    where
        action_flg = 'I') t3 on
    t1.yesterday_dt = t3.dt
    and t1.lot_min_price = t3.initial_price;
"""

q_stats_2gis = """
select
    create_dttm::date,
    count(*) as media_cnt_2gis,
    count(distinct lot_id) as lots_cnt_2gis
from
    etl.lot_media_2gis
    where create_dttm::date = now()::date
group by
    create_dttm::date
"""

q_stats_dom_rf = """
select
    count(*) as lots_cnt_dom_rf
from
    etl.lot_data_dom_rf
    where created_dttm::date = now()::date - interval '1' day
"""
# ------------------------------------------------------------------------------
# маска рассылки
# ------------------------------------------------------------------------------

message_msk = """Статистика за {yestertay_dt}:\n
- Загружено новых лотов: {new_lots_cnt} шт., из них валидные: {new_lots_valid_cnt} шт.
- Удалено лотов: {deleted_lots_cnt} шт.
- Обновлено лотов: {updated_lots_cnt} шт.
- Максимальная стоимость валидного {lot_id_max_link}: {lot_max_price} млн.
- Минимальная стоимость валидного {lot_id_min_link}: {lot_min_price} млн.

Статистика 2gis:\n
- Загружено изображений: {media_cnt_2gis} шт.
- Обогащено лотов: {lots_cnt_2gis} шт.

Статистика dom-rf:\n
- Обогащено лотов: {lots_cnt_dom_rf} шт.
"""

message_msk_short = """Статистика за {yestertay_dt}:\n
\n
- Загружено новых лотов: {new_lots_cnt} шт. \n
- Удалено лотов: {deleted_lots_cnt} шт. \n
- Обновлено лотов: {updated_lots_cnt} шт. \n
"""

def create_link(lot_id, word, url = "https://xn--p1abd.xn----etbpba5admdlad.xn--p1ai/lot/"):
    url = url + str(lot_id)
    return """<a href='{url}'>{word}</a>""".format(url = url, word = word)

def calculate_stats_from_pg(**kwargs):
    conn = create_engine(conn_string, connect_args={"options": "-c timezone=Europe/Moscow"})
    print("""Start calculate statistics for yesterday""")
    df_stats = pd.read_sql_query(sql = q_stats, con = conn).fillna(0)
    df_stats_2gis = pd.read_sql_query(sql = q_stats_2gis, con = conn).fillna(0)
    df_stats_dom_rf = pd.read_sql_query(sql = q_stats_dom_rf, con = conn).fillna(0)

    if df_stats['new_lots_cnt'].values[0]>0:
        stats_dict = {col:df_stats[col].values[0] for col in df_stats.columns}
        stats_2gis_dict = {col:df_stats_2gis[col].values[0] for col in df_stats_2gis.columns}
        stats_dom_rf_dict = {col:df_stats_dom_rf[col].values[0] for col in df_stats_dom_rf.columns}

        message_tg = message_msk.format(yestertay_dt = date.strftime(stats_dict['yesterday_dt'], '%Y-%m-%d'),
                            new_lots_cnt = stats_dict['new_lots_cnt'], 
                            new_lots_valid_cnt = stats_dict['new_lots_valid_cnt'], 
                            deleted_lots_cnt = stats_dict['deleted_lots_cnt'],
                            updated_lots_cnt = stats_dict['updated_lots_cnt'],
                            lot_max_price = round(stats_dict['lot_max_price']/1e6, 2),
                            lot_min_price = round(stats_dict['lot_min_price']/1e6, 2), 
                            lot_id_max_link = create_link(stats_dict['lot_id_max_price'], word = 'лота'),
                            lot_id_min_link = create_link(stats_dict['lot_id_min_price'], word = 'лота'),
                            media_cnt_2gis = stats_2gis_dict['media_cnt_2gis'],
                            lots_cnt_2gis = stats_2gis_dict['lots_cnt_2gis'],
                            lots_cnt_dom_rf = stats_dom_rf_dict['lots_cnt_dom_rf']
                            )
        ti = kwargs['ti']
        ti.xcom_push(value=message_tg, key='message_tg')
    else:
        df_stats = df_stats[['yesterday_dt',  'new_lots_cnt', 'updated_lots_cnt', 'deleted_lots_cnt']].copy()
        stats_dict = {col:df_stats[col].values[0] for col in df_stats.columns}
        message_tg = message_msk_short.format(yestertay_dt = date.strftime(stats_dict['yesterday_dt'], '%Y-%m-%d'),
                            new_lots_cnt = stats_dict['new_lots_cnt'], 
                            deleted_lots_cnt = stats_dict['deleted_lots_cnt'],
                            updated_lots_cnt = stats_dict['updated_lots_cnt']
                            )
        ti = kwargs['ti']
        ti.xcom_push(value=message_tg, key='message_tg')


def print_xcom(**kwargs):
    ti = kwargs['ti']
    print('The value is: {}'.format(
        ti.xcom_pull(task_ids='calculate_stats_from_pg')
    ))
    print('')
    print(list(kwargs.keys()))
    

with DAG(
        dag_id='stats_to_tg',
        start_date = datetime(year = 2023, month = 3, day = 9, hour = 10, minute = 30, tzinfo=LOCAL_TZ),
        schedule_interval='30 10 * * *',
        catchup=False,
        default_args = {
           'owner':'d.kakurin',
       },
        tags=['tg'],
) as dag:
    start_task = EmptyOperator(
        task_id="start_task"
    )

    get_stats_from_db = PythonOperator(
        task_id='calculate_stats_from_pg',
        python_callable=calculate_stats_from_pg
    )

    print_xcom = PythonOperator(
        task_id='printer',
        python_callable=print_xcom
    )

    end_task = EmptyOperator(
        task_id="end_task"
    )

    send_data_to_tg = TelegramOperator(
        task_id='send_data_to_tg',
        token=TG_BOT_TOKEN, 
        chat_id=TG_CHAT_ID, 
        text='{{ ti.xcom_pull(task_ids="calculate_stats_from_pg", key="message_tg") }}'
    )

start_task >> get_stats_from_db >> print_xcom >> send_data_to_tg >> end_task